package com.talent.service.iot.mqtt.service;

import com.talent.service.iot.mqtt.model.request.TlMqttPubRecReq;
import com.talent.service.iot.mqtt.model.request.TlMqttPubRelReq;
import com.talent.service.iot.mqtt.model.request.TlMqttPublishReq;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName：DefaultMqttMsgServiceImpl
 * @Author: hszhou
 * @Date: 2025/1/8 13:51
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
public class LocalMqttMsgServiceImpl implements IMqttMsgService{

    /**
     * broker用于存储本地的消息
     */
    public static final ConcurrentHashMap<String,ConcurrentHashMap<String,TlMqttPublishReq>> PUBLISH_MAP=new ConcurrentHashMap<>();


    public static final ConcurrentHashMap<String, ConcurrentHashMap<String,TlMqttPubRelReq> > PUBREL_MAP=new ConcurrentHashMap<>();


    public static final ConcurrentHashMap<String, ConcurrentHashMap<String,TlMqttPubRecReq> > PUBREC_MAP=new ConcurrentHashMap<>();



    private final static AtomicInteger IDENTIFIER_MESSAGE = new AtomicInteger(0);

    @Override
    public  Mono<Void> savePublishMessage(String clientId, int messageId, TlMqttPublishReq req) {

        PUBLISH_MAP.compute(clientId, (k, v) -> {
            if (v == null) {
                v = new ConcurrentHashMap<>();
            }
            v.computeIfAbsent(String.valueOf(messageId),key-> req);
            return v;
        });

        return Mono.empty();
    }

    @Override
    public  Mono<Void> clearPublishMessage(String clientId, int messageId) {

        PUBLISH_MAP.getOrDefault(clientId,new ConcurrentHashMap<>())
                        .remove(String.valueOf(messageId));
        return Mono.empty();
    }

    @Override
    public  Mono<Void> clearPublishMessage(String clientId) {
        PUBLISH_MAP.remove(clientId);
        return Mono.empty();
    }

    @Override
    public Mono<TlMqttPublishReq> findPublishMessage(String clientId, int messageId) {
        return Mono.justOrEmpty(PUBLISH_MAP.getOrDefault(clientId,new ConcurrentHashMap<>()).get(String.valueOf(messageId)));
    }

    @Override
    public Flux<TlMqttPublishReq> findPublishMessage(String clientId) {
        return null;
    }

    @Override
    public  Mono<Void> savePubRelMessage(String clientId, int messageId, TlMqttPubRelReq req) {

        PUBREL_MAP.compute(clientId, (k, v) -> {
            if (v == null) {
                v = new ConcurrentHashMap<>();
            }
            v.computeIfAbsent(String.valueOf(messageId),key-> req);
            return v;
        });

        return Mono.empty();
    }



    @Override
    public  Mono<Void> clearPubRelMessage(String clientId, int messageId) {
        PUBREL_MAP.getOrDefault(clientId,new ConcurrentHashMap<>())
                .remove(String.valueOf(messageId));
        return Mono.empty();
    }

    @Override
    public  Mono<Void> clearPubRelMessage(String clientId) {
        PUBREL_MAP.remove(clientId);
        return Mono.empty();
    }

    @Override
    public Mono<TlMqttPubRelReq> findPubRelMessage(String clientId, int messageId) {
        return Mono.justOrEmpty(PUBREL_MAP.getOrDefault(clientId,new ConcurrentHashMap<>()).get(String.valueOf(messageId)));
    }

    @Override
    public Flux<TlMqttPubRelReq> findPubRelMessage(String clientId) {
        return null;
    }

    @Override
    public Integer getMessageId() {
        return IDENTIFIER_MESSAGE.getAndIncrement();
    }

    @Override
    public Mono<Void> savePubRecMessage(String clientId, int messageId, TlMqttPubRecReq req) {
        PUBREC_MAP.compute(clientId, (k, v) -> {
            if (v == null) {
                v = new ConcurrentHashMap<>();
            }
            v.computeIfAbsent(String.valueOf(messageId),key-> req);
            return v;
        });

        return Mono.empty();
    }

    @Override
    public Mono<Void> clearPubRecMessage(String clientId, int messageId) {

        PUBREC_MAP.getOrDefault(clientId,new ConcurrentHashMap<>())
                .remove(String.valueOf(messageId));
        return Mono.empty();
    }
}
